package us.codecraft.webmagic.pipeline;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableNotFoundException;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;
import org.jsoup.nodes.Element;

import us.codecraft.webmagic.ResultItems;
import us.codecraft.webmagic.Task;
import us.codecraft.webmagic.utils.FilePersistentBase;

public class HbasePipeline extends FilePersistentBase implements Pipeline {
	Configuration cfg = null;
	String tableName;
	HTable table;
	String url;
	HashSet<String> hs_tagNames = new HashSet<String>();
	static HBaseAdmin admin;
	
	public HbasePipeline (String tableName) throws Exception {
		cfg = HBaseConfiguration.create();    
		cfg.set("hbase.zookeeper.quorum", "localhost");
		cfg.set("hbase.zookeeper.property.clientPort", "2181");	    
		
		this.tableName = tableName;
		admin = new HBaseAdmin(cfg);
		createTable("title");
		table = new HTable(cfg, tableName);
	}
	
	@Override
	public void process(ResultItems resultItems, Task task) {
		// TODO Auto-generated method stub
		url = resultItems.getRequest().getUrl();
		
		Document doc = Jsoup.parse(resultItems.getAll().entrySet().toString());
		
		for (Element el : doc.getAllElements()) {
			if (!el.tagName().matches("script|link|#root|html|iframe|form")) hs_tagNames.add(el.tagName());
		} 
		
		try {
			TagToHbase (doc);
		} catch (TableNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

	}

	public void createTable(String ... colFamilies) throws Exception {
        if (admin.tableExists(tableName)) {
            System.out.println("table   Exists!!!");
        }
        else{
            HTableDescriptor tableDesc = new HTableDescriptor(tableName);
            for (String colFam : colFamilies) tableDesc.addFamily(new HColumnDescriptor(colFam));
            admin.createTable(tableDesc);
            System.out.println("create table ok .");
        }
	}
	
	public void TagToHbase (Document doc) throws TableNotFoundException, IOException {
		List<Put> put = new ArrayList<Put>();
		
		for (String s : hs_tagNames) {
			if (!admin.getTableDescriptor(tableName.getBytes()).hasFamily(s.getBytes())) {
				admin.disableTable(tableName);
				HColumnDescriptor cf1 = new HColumnDescriptor(s);
				admin.addColumn(tableName, cf1);
				admin.enableTable(tableName);
			}
			
			for (Element el : doc.getElementsByTag(s)) {
				
				put.add(new Put(Bytes.toBytes(url)).add(
										Bytes.toBytes(s),//column family
										Bytes.toBytes(Integer.toString(doc.toString().indexOf(el.toString()))),//column qualifier
										Bytes.toBytes(el.attr("href")+";"+el.text())//value
										)
						);
			}
		}
		
		table.put(put);
	}
}